use crate::{
    cmd, config,
    data::{self},
    guard,
    ws::{self, WsRequest, WsResponse, WsResult},
};
use actix_codec::Framed;
use actix_web::ResponseError;
use awc::{
    error,
    ws::{Codec, Frame, Message},
    BoxedSocket, ClientResponse,
};
use bytes::Bytes;
use bytestring::ByteString;
use futures::{SinkExt as _, StreamExt as _};
use std::{error::Error, io::ErrorKind, time::Duration};
use tokio::{
    select,
    sync::mpsc::{self, UnboundedSender},
    time::{self},
};
use tokio_stream::wrappers::UnboundedReceiverStream;
use tokio_util::sync::CancellationToken;

pub const INTERVAL: Duration = Duration::from_secs(1);
pub const START_AFTER: Duration = Duration::from_secs(3);

// 重置后，要间隔多久才开始投票
// pub const RESET_INTERVAL: Duration = Duration::from_secs(3);

// const LOCK_TIMEOUT: Duration = Duration::from_secs(3);
// fn get_path_from_sn(sn: HashMap<String, String>) -> Vec<String> {
//     let cws = config::get_server().and_then(|s| s.watcher);
//     let mut paths = Vec::new();
//     if let Some(cws) = cws {
//         // 监听所有路径
//         for (srv_name, cw) in cws {
//             let value = sn.get(&srv_name);
//             for (namespace, cwe) in cw {
//                 // 基础路径
//                 if let Some(value) = value {
//                     if namespace == *value {
//                         paths.push(cwe.root);
//                     }
//                 }
//             }
//         }
//     }
//     paths
// }

/// 处理 ws响应的各种事件
fn handle_rsp_evt(v: WsResponse, server_id: usize) {
    let myid = config::get_server_id();

    // 只处理心跳事件
    // if server_id == myid {
    //     if let Some(evt) = v.evt {
    //         match evt {
    //             // 心跳完成
    //             ws::WsEvent::HB => {
    //                 // log::debug!("Heartbeat::lock::{}", lock.is_locked());
    //                 let sga = lock.try_lock().unwrap();
    //                 log::debug!("[{myid}=>{server_id}] - [{leader}] - {:?} Heartbeat Ok", sga.role);
    //             }
    //             // ws::WsEvent::HL => {
    //             //     let mut loc = lock.try_lock().unwrap();
    //             //     // 重置
    //             //     loc.reset_with_vote();
    //             //     // 通知其他等待的线程
    //             //     cvar.notify_one();
    //             // }
    //             _ => {
    //                 log::debug!("[{myid}=>{server_id}] - [{leader}] - Unsupported events: {:?}", &evt);
    //             }
    //         }
    //     }

    //     return;
    // }

    // 接收到其他数据
    // 数据分析

    if let Some(evt) = v.evt {
        match evt {
            ws::WsEvent::BF => {
                // 收到广播的文件
                if v.dat.is_none() {
                    return;
                }
                let vcd = v.vcd.unwrap();
                // 收到广播数据
                /*Mutex*/
                {
                    let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                    let mut loc = lock.lock();
                    // Acquired Lock: ReadWrite
                    // 同步数据
                    loc.vcd.sync_with(&vcd);
                    // 通知其他等待的线程
                    cvar.notify_all();
                    // Released Lock
                }

                let dat = v.dat.unwrap();
                log::trace!("dat::{:?}", dat);
                for (b_srv_name, data) in dat {
                    for (srv_name, en) in config::get_registry_services() {
                        if b_srv_name != srv_name {
                            continue;
                        }
                        // enable_broadcast => false
                        if let Some(enable_broadcast) = en.enable_broadcast {
                            if !enable_broadcast {
                                continue;
                            }
                        } else {
                            continue;
                        }
                        let rbe = match en.broadcast {
                            Some(broadcast) => broadcast,
                            None => continue,
                        };
                        // enable_broadcast => false ... end

                        if let Some(output) = rbe.output {
                            crate::file::write_broadcast_output(
                                &b_srv_name,
                                &output,
                                en.root,
                                &data,
                            )
                        }
                    }
                }
            }
            ws::WsEvent::TV => {
                if v.vcd.is_none() {
                    return;
                }

                let vcd = v.vcd.unwrap();

                // Follower下线
                log::info!("Follower {server_id} off-lined, Nothing to do");

                /*LOCK*/
                {
                    let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                    let mut loc = lock.lock();
                    // Acquired Lock: ReadWrite
                    // 通知其他节点已离线，并非当前server_id离线
                    loc.offline(vcd.myid);
                    // 通知其他等待的线程
                    cvar.notify_all();
                    // Released Lock
                }
            }

            ws::WsEvent::LV => {
                // 投票
                match v.res {
                    // 投票成功
                    WsResult::Ok => {
                        if v.vcd.is_none() {
                            return;
                        }

                        let vcd = v.vcd.unwrap();
                        let leader = vcd.leader;
                        log::debug!("[{myid}=>{server_id}] - [{leader}] - {}", v.msg);
                        log::info!("Successfully voted to {server_id}");

                        cleanup_scratch(Some(server_id));
                    }
                    WsResult::Failed => {
                        if v.vcd.is_none() {
                            return;
                        }

                        let vcd = v.vcd.unwrap();
                        let leader = vcd.leader;

                        // 投票失败
                        log::debug!("[{myid}=>{server_id}] - [{leader}] - {}", v.msg);

                        take_back_from_scratch(Some(server_id));
                    }
                    // ERROR
                    _ => {}
                }
            }

            ws::WsEvent::FA => {
                // leader选出后，会广播到这个事件

                if v.vcd.is_none() {
                    log::error!("evt::FA, vcd is none");
                    return;
                }
                let vcd = v.vcd.unwrap();
                let leader = vcd.leader;
                log::debug!("[{myid}=>{server_id}] - [{leader}] - {}", v.msg);

                // 收到广播数据
                /*Mutex*/
                {
                    let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                    let mut loc = lock.lock();
                    // Acquired Lock: ReadWrite

                    // 同步数据
                    loc.vcd.sync_with(&vcd);
                    loc.retrieve();
                    log::info!("Received broadcast from Leader, Set status={:?}->Following, myid={myid}, leader={leader}", loc.status);
                    loc.status = data::Status::Following;

                    // 数据同步完成后，支持自动切换功能
                    loc.enable_auto_switchover();

                    // 通知其他等待的线程
                    cvar.notify_all();
                    // Released Lock
                }

                // 切换为follower
                cmd::on_switch_to_follower();
            }

            ws::WsEvent::LA => {
                // 这个事件为需要重新选举leader
                if v.vcd.is_none() {
                    return;
                }
                let vcd = v.vcd.unwrap();

                /*Mutex*/
                {
                    let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                    let mut loc = lock.lock();
                    // Acquired Lock: ReadWrite

                    // 同步数据
                    log::debug!("Leader off-lined from LA event, Start a new round of voting with {server_id}");
                    loc.reset();
                    loc.vcd.sync_with(&vcd);

                    // 通知其他等待的线程
                    cvar.notify_all();
                    // Released Lock
                }
            }

            ws::WsEvent::RV => {
                // 应某节点要求，重新选leader
                if v.vcd.is_none() {
                    return;
                }
                let vcd = v.vcd.unwrap();

                /*Mutex*/
                {
                    let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                    let mut loc = lock.lock();
                    // Acquired Lock: ReadWrite

                    // 同步数据
                    log::info!("Leader Election from RV event, Start a new round of voting with {server_id}");
                    loc.reset();
                    loc.vcd.sync_with(&vcd);

                    // 通知其他等待的线程
                    cvar.notify_all();
                    // Released Lock
                }
            }

            _ => {}
        }
    }
}

/// 处理请求发送事件
fn post_req_evt(
    server_id: usize,
    tx: UnboundedSender<WsRequest>,
) -> Result<(), tokio::sync::mpsc::error::SendError<WsRequest>> {
    let leader;
    let cloned_vcd;
    let has_leader;
    let is_pending;
    // let is_voted;
    let is_poll_from;
    let is_poll_failed;
    let is_poll_to;

    let has_scratch_poll;
    let is_auto_switchover;
    {
        let &(ref lock, ref _cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
        let loc = lock.lock();
        // Acquired Lock: ReadOnly
        leader = loc.vcd.leader;
        cloned_vcd = loc.get_vcd();
        has_leader = loc.has_leader();
        is_pending = loc.is_pending();
        // is_voted = loc.is_voted(&server_id);
        is_poll_from = loc.is_poll_from(&server_id);
        is_poll_failed = loc.is_poll_failed(&server_id);
        is_poll_to = loc.is_poll_to(&server_id);

        has_scratch_poll = loc.has_scratch_poll();
        is_auto_switchover = loc.is_auto_switchover();
        // Released Lock
    }

    let event = if !has_leader {
        // 如果已经放弃候选人的角色，则后续无需参与投票
        // 通过判断投票来源，避免重复投票
        if is_pending {
            // 票已投出，等待选举
            log::info!("Voted and Pending, wait {} secs", INTERVAL.as_secs());
            // 发送心跳
            ws::WsEvent::PP
        } else if is_poll_from {
            log::info!("Voted from {server_id}, wait {} secs", INTERVAL.as_secs());
            // 发送心跳
            ws::WsEvent::PP
        } else if is_poll_to {
            log::info!("Voted to {server_id}, wait {} secs", INTERVAL.as_secs());
            // 发送心跳
            ws::WsEvent::PP
        } else if is_poll_failed {
            log::info!(
                "Voted failed for {server_id}, wait {} secs",
                INTERVAL.as_secs()
            );
            // 发送心跳
            ws::WsEvent::PP
        } else if has_scratch_poll {
            // 票未投出，在投票的过程中
            log::info!(
                "Poll scratching for {server_id}, wait {} secs",
                INTERVAL.as_secs()
            );
            // 如果有暂存，则等待
            ws::WsEvent::PP
        } else {
            // 投票
            // 如果 server_id 在 Attendeess列表中，则投票
            let connected;
            {
                let &(ref lock, ref _cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                let loc = lock.lock();
                // Acquired Lock: ReadOnly
                connected = loc.attendees.contains_key(&server_id)
                // Released Lock
            }

            if !connected {
                log::debug!("The {server_id} not in attendees");
                return Ok(());
            }

            ws::WsEvent::LV
        }
    } else if leader == server_id {
        // 自动切换功能
        if is_auto_switchover {
            log::info!("Enable AUTO-SWITCHOVER Future, Required new election and Send to Leader {server_id}");
            {
                let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                let mut loc = lock.lock();
                // Acquired Lock: ReadWrite
                loc.disable_auto_switchover();
                // Released Lock
                cvar.notify_all();
            }
            ws::WsEvent::RV
        } else {
            // 只从leader同步数据
            // ws::WsEvent::DS
            // 需要等待leader广播数据，而不是主动向leader取数据，因此需要保持连接
            log::debug!("Waiting for broadcast events");
            ws::WsEvent::PP
        }
    } else {
        // 其他节点发送Ping
        ws::WsEvent::PP
    };

    let data = WsRequest {
        evt: event.clone(),
        vcd: cloned_vcd,
    };

    match event {
        ws::WsEvent::LV => {
            // 票已投出，暂存票数
            {
                let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                let mut loc = lock.lock();
                // Acquired Lock: ReadWrite
                loc.scratch_for(server_id);
                // Released Lock
                cvar.notify_all();
            }
        }
        _ => {}
    }

    // 发送数据
    let res = tx.send(data);
    // drop(tx);
    res
}

// 如果网络异常或投票失败后，应该从暂存中取回票
fn take_back_from_scratch(failed_server_id: Option<usize>) {
    let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
    let mut loc = lock.lock();
    // Acquired Lock: ReadWrite
    if let Some(id) = failed_server_id {
        log::info!(
            "Voting failed to {id}, Take back my Poll {}",
            loc.scratch_poll
        );
        // 投票失败处理
        let p: usize = loc.scratch_poll;
        loc.voting_failed(id, p);
    }
    loc.retrieve();
    cvar.notify_all();
    // Released Lock
}

// 投票成功处理
fn cleanup_scratch(to_server_id: Option<usize>) {
    let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
    let mut loc = lock.lock();
    // Acquired Lock: ReadWrite
    if let Some(id) = to_server_id {
        // 更新投票去向
        loc.poll_to_with_scratch(id);
    }
    // 状态即将转换为Following
    loc.status = data::Status::Pending;
    // 投出后，清除暂存票数
    loc.clean_scratch();
    // loc.pollout();
    // 通知其他等待的线程
    cvar.notify_all();
    // Released Lock
}

// 获取websocket连接
async fn get_ws_connection(
    url: &str,
    server_addr: &str,
    server_port: &u16,
    app_key: &str,
    app_secret: &str,
    server_id: usize,
) -> (ClientResponse, Framed<BoxedSocket, Codec>) {
    let client_connect_retry_secs = config::get_client_connect_retry_wait();

    // let my_local_ip = match local_ip_address::local_ip() {
    //     Ok(local_ip) => format!("{:?}", local_ip),
    //     Err(_) => "127.0.0.1".to_owned()
    // };

    let json_members = match serde_json::to_string(&config::get_members()) {
        Ok(json) => json,
        Err(e) => {
            log::error!("config::get_members to json strings failed, cause: {e}");
            String::new()
        }
    };

    loop {
        let r = awc::Client::new()
            .ws(url)
            .header(guard::im::HEADER_APP_KEY, app_key)
            .header(guard::im::HEADER_APP_SECRET, app_secret)
            .header(guard::im::HEADER_KEY_MEMBERS, json_members.clone())
            .connect()
            .await;

        match r {
            Ok((rsp, ws)) => {
                /*Mutex*/
                {
                    let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                    // Acquired Lock: ReadWrite
                    let mut loc = lock.lock();
                    loc.online(server_id);
                    cvar.notify_all();
                    // Released Lock
                }
                return (rsp, ws);
            }
            // Err(error::WsClientError::InvalidResponseStatus(status)) => {
            //     println!("error::WsClientError::InvalidResponseStatus: {:?}", status);
            // }
            // Err(error::WsClientError::InvalidUpgradeHeader) => {
            //     println!("error::WsClientError::InvalidUpgradeHeader");
            // }
            // Err(error::WsClientError::InvalidConnectionHeader(value)) => {
            //     println!("error::WsClientError::InvalidConnectionHeader:{:?}", value);
            // }
            // Err(error::WsClientError::MissingConnectionHeader) => {
            //     println!("error::WsClientError::MissingConnectionHeader");
            // }
            // Err(error::WsClientError::Protocol(ws_protocol_error)) => {
            //     println!("error::WsClientError::Protocol: {:?}", ws_protocol_error);
            //     match ws_protocol_error {
            //         error::WsProtocolError::Io(err) => {
            //             if err.kind() == ErrorKind::HostUnreachable {
            //                 // No route to host
            //                 // 启用防火墙
            //                 println!("No route to host...")
            //             }
            //             println!("error::WsProtocolError::Io: {:?}", err);
            //         }
            //         _ => {}
            //     }
            // }
            Err(error::WsClientError::SendRequest(send_request_error)) => {
                // println!(
                //     "error::WsClientError::SendRequest: {:?}",
                //     send_request_error
                // );
                match send_request_error {
                    error::SendRequestError::Connect(conn_err) => {
                        match conn_err {
                            error::ConnectError::Io(err) => {
                                match err.kind() {
                                    ErrorKind::HostUnreachable | ErrorKind::TimedOut => {
                                        // 网络不开放
                                        // 连接超时
                                        log::error!("Failed to connect to host: {err} [{server_id}] -> {server_addr}:{server_port}");
                                        // 对于网络不开放的情况，且当前没有header时，则直接抛出错误信息
                                        // {
                                        //     let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                                        //     let loc = lock.lock();
                                        //     // Acquired Lock: ReadOnly
                                        //     if !loc.has_leader() {
                                        //         log::error!("Vote failed");
                                        //     }
                                        //     // Released Lock
                                        //     cvar.notify_all();
                                        // }
                                    }
                                    // ErrorKind::ConnectionRefused => {
                                    //     // 服务未启动
                                    //     log::error!("Failed to connect to host: {err} [{server_id}] -> {server_addr}:{server_port}");
                                    //     disconnect(server_id);
                                    // }
                                    // ErrorKind::TimedOut => {
                                    //     // 服务未启动
                                    //     log::error!("{err} [{server_id}] -> {server_addr}:{server_port}");
                                    // }
                                    _ => {
                                        log::error!("Failed to connect to host: {err} [{server_id}] -> {server_addr}:{server_port}");
                                        disconnect(server_id);
                                    }
                                }
                            }
                            error::ConnectError::Timeout => {
                                // 连接超时，网络不开放
                                log::error!("Failed to connect to host: Connection timed out (os error 110) [{server_id}] -> {server_addr}:{server_port}");
                            }
                            _ => {
                                log::error!("Failed to connect to host: {conn_err} [{server_id}] -> {server_addr}:{server_port}");
                                disconnect(server_id);
                            }
                        }
                    }
                    _ => {
                        log::error!("Failed to connect to host: {send_request_error} [{server_id}] -> {server_addr}:{server_port}");
                        disconnect(server_id);
                    }
                }
            }
            Err(e) => {
                log::error!("{e} [{server_id}] -> {server_addr}:{server_port}");

                disconnect(server_id);
            }
        }

        log::debug!(
            "Connection not established, wait {client_connect_retry_secs} secs for retry again"
        );
        time::sleep(Duration::from_secs(client_connect_retry_secs as u64)).await;
    }
}

// 等待10秒
fn disconnect(server_id: usize) {
    /*LOCK*/
    {
        let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
        let mut loc = lock.lock();
        // Acquired Lock: ReadWrite
        if loc.is_leader(server_id) {
            // 如果连接失败的是leader
            // 重置，开始投票
            log::info!("Leader Disconnect, Status Reset to Looking");
            loc.reset();
        }
        // 该节点不活跃
        loc.offline(server_id);
        cvar.notify_all();
        // Released Lock
    }
}

pub struct ClientHandler {
    server_addr: String,
    server_port: u16,
    server_id: usize,
    url: String,
}

impl ClientHandler {
    pub fn new(server_addr: String, server_port: u16, server_id: usize) -> Self {
        let myid = config::get_server_id();
        ClientHandler {
            url: format!("ws://{server_addr}:{server_port}/im/chat/{server_id}/{myid}"),
            server_addr,
            server_port,
            server_id,
        }
    }

    // 启动
    pub async fn start(&mut self, app_key: &str, app_secret: &str) {
        let server_id = self.server_id;
        let (_rsp, mut ws) = get_ws_connection(
            &self.url,
            &self.server_addr,
            &self.server_port,
            app_key,
            app_secret,
            server_id,
        )
        .await;
        log::info!(
            "Connection established [{server_id}] -> {}:{}",
            &self.server_addr,
            &self.server_port
        );

        // token不能放在new函数初始化！！！！！！！
        // 正向:从定时器发送数据,并到ws接收数据
        let (tx, _rx) = mpsc::unbounded_channel::<WsRequest>();
        let mut rx = UnboundedReceiverStream::new(_rx);

        // token不能放在new函数初始化！！！！！！！
        // 反向:从ws接收,并传输到定时器
        let (rtx, _rrx) = mpsc::unbounded_channel::<WsResponse>();
        let mut rrx = UnboundedReceiverStream::new(_rrx);

        // token不能放在new函数初始化！！！！！！！
        let token = CancellationToken::new();
        let cloned_token = token.clone();

        // 启动定时发送任务
        let handle = tokio::spawn(async move {
            let mut intv = time::interval_at(tokio::time::Instant::now() + START_AFTER, INTERVAL);
            let start_secs = START_AFTER.as_secs();
            let intv_secs = START_AFTER.as_secs();
            log::info!("Sender#{server_id} inited, start after {start_secs} secs, running every {intv_secs} secs");

            loop {
                select! {
                    _ = cloned_token.cancelled() => {
                        log::info!("Sender#{server_id} cancelled");
                        take_back_from_scratch(None);
                        break;
                    },
                    Some(v) = rrx.next() => handle_rsp_evt(v, server_id),
                    _ = intv.tick() => {
                        // 每隔一段时间运行一次
                        if let Err(e) = post_req_evt(server_id, tx.clone()) {
                            // TX send failed, cause: channel closed
                            // 连接不上Leader时？清空自己的leader信息，并重置
                            disconnect(server_id);
                            log::error!("TX send to {server_id} failed, cause: {e}");
                            take_back_from_scratch(None);
                            break;
                        }
                        log::debug!("TX sent to {server_id}, wait {} secs", INTERVAL.as_secs());
                    }
                }
            }
        });

        loop {
            tokio::select! {
                // 接收信息
                Some(payload) = ws.next() => {
                    let payload = match payload {
                        Ok(f) => f,
                        Err(_) => {
                            return;
                        }
                    };

                    match payload {
                        Frame::Text(msg) => {
                            // 接收数据解析
                            let raw = String::from_utf8_lossy(msg.as_ref()).to_string();
                            log::trace!("Recv data from [{server_id}] <- {raw}");

                            let v = match WsResponse::from_str(&raw) {
                                Ok(v) => v,
                                Err(e) => {
                                    log::error!("WsParseError: `{raw}` is not valid JSON, cause: {e}");
                                    return;
                                }
                            };
                            // 状态码判断
                            match v.res {
                                WsResult::Error => {
                                    log::error!("The server {server_id} returned an error: {}", v.msg);
                                    return;
                                }
                                _ => {
                                }
                            }

                            // 将数据传回定时器
                            if let Err(e) = rtx.send(v) {
                                log::error!("RTX send to {server_id} failed, cause: {e}");
                            }
                        }
                        Frame::Ping(msg) => {
                            if let Err(e) = ws.send(Message::Pong(msg)).await {
                                log::error!("Ping failed with {server_id}, cause: {e}");
                            }
                        }
                        Frame::Pong(msg) => {
                            let raw = String::from_utf8_lossy(msg.as_ref()).to_string();
                            log::debug!("Pong with {server_id}: {}", raw);
                        }
                        Frame::Continuation(msg) => {
                            log::debug!("Continuation with {server_id}: {:?}", msg);
                        }
                        Frame::Close(msg) => {
                            log::debug!("Connection closed with {server_id}, cause: {}", msg.unwrap().description.unwrap());
                            break;
                        }
                        Frame::Binary(msg) => {
                            log::debug!("Binary with {server_id}: {:?}", msg);
                        }
                    }
                },

                // 定时发送信息
                Some(req) = rx.next() => {
                    log::trace!("RX recv data: {:?}", req);

                    let mut ping = false;
                    let buf: ByteString;
                    let msg = match req.evt {
                        // Ping Pong
                        ws::WsEvent::PP => {
                            ping = true;
                            let hello = concat!(env!("CARGO_PKG_NAME"), "-", env!("CARGO_PKG_VERSION"));
                            buf = ByteString::from_static(hello);
                            Message::Ping(Bytes::from_static(hello.as_bytes()))
                        }
                        _ => {
                            buf = req.to_bytestr();
                            Message::Text(buf.clone())
                        }
                    };

                    match ws.send(msg).await {
                        Ok(_r) => {
                            if ping {
                                log::debug!("Ping with {server_id}: {buf}");
                            } else {
                                log::trace!("Data: {buf}");
                                log::info!("Ping data sent to {server_id}");
                            }
                        },
                        Err(actix_http::ws::ProtocolError::Io(e)) => {
                            // I/O error: Broken pipe (os error 32)
                            // Disconnect
                            log::trace!("Data: {buf}");
                            log::error!("Send failed to {server_id}, cause: I/O error: {e}");
                            take_back_from_scratch(None);
                            token.cancel();
                            break;
                        },
                        // Err(actix_http::ws::ProtocolError::UnmaskedFrame) => {},
                        // Err(actix_http::ws::ProtocolError::MaskedFrame) => {},
                        // Err(actix_http::ws::ProtocolError::BadOpCode) => {},
                        // Err(actix_http::ws::ProtocolError::Overflow) => {},
                        // Err(actix_http::ws::ProtocolError::ContinuationNotStarted) => {},
                        // Err(actix_http::ws::ProtocolError::ContinuationStarted) => {},
                        Err(e) => {
                            log::trace!("server_id={server_id}, status_code={}, source={:?}, error_response={:?}", e.status_code(), e.source(), e.error_response());
                            log::debug!("Data: {buf}");
                            log::error!("Send failed to {server_id}, cause: {e}");
                            take_back_from_scratch(None);
                        }

                    }
                },

                else => {
                    // 未匹配
                }
            }
        }
        handle.await.unwrap();

        // drop
        log::trace!(
            "Connection closed [{server_id}] -> {}:{}",
            &self.server_addr,
            &self.server_port
        );
    }
}
